package org.apache.lorry.zookeeper.cluster;

import com.alibaba.fastjson2.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.Serializable;

@Slf4j
public class WorkTask implements Runnable, Serializable {
    private String taskName;

    private String workerName;
    private ZooKeeper zk;

    public WorkTask(String taskName, String workerName, ZooKeeper zk) {
        this.taskName = taskName;
        this.workerName = workerName;
        this.zk = zk;
    }

    AsyncCallback.DataCallback getTaskCallback = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    getTaskData();  //进行重试
                    break;
                case OK:
                    String dataStr = new String(data);
                    Task task = JSONObject.parseObject(dataStr, Task.class);
                    task.execute(); //执行任务
                    zk.create(ZooConstant.STATUS_PATH + "/" + taskName, "done".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, statusDataCallback, null);   //创建状态完成节点
                    zk.delete(getTaskPath(), -1, deleteTaskCallback, null);     //删除work下任务节点
                    break;
                default:
                    System.out.println("Something went wrong here, " +
                            KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    AsyncCallback.VoidCallback deleteTaskCallback = new AsyncCallback.VoidCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    zk.delete(getTaskPath(), -1, deleteTaskCallback, null); //再删一次
                    break;
                case OK:
                    log.info("Successfully deleted {}", path);
                    break;
                default:
                    log.error("Something went wrong here, " +
                            KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    AsyncCallback.StringCallback statusDataCallback = new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    zk.create(path, "done".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, statusDataCallback, null);
                    break;
                case OK:
                    log.info("Status node is created successfully: {}", path);
                    break;
                case NODEEXISTS:
                    log.info("Status node is already exists: {}", path);
                    break;
                default:
                    log.error("Something went wrong here, " +
                            KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    private String getTaskPath() {
        return ZooConstant.ASSIGN_PATH + "/" + workerName + "/" + taskName;
    }

    private void getTaskData() {
        zk.getData(getTaskPath(), false, getTaskCallback, null);
    }



    @Override
    public void run() {
        getTaskData();
    }
}
